package com.itcast.report

import com.itcast.beans.ReportMediaAnalysis
import com.itcast.utils.{ConfigHandler, MysqlHandler, RptKpiTools}
import org.apache.commons.lang3.StringUtils
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext


object ReqMediaAnalysis {
  def main(args: Array[String]): Unit = {
    //创建sparkConf
    val sparkConf = new SparkConf()
      .setAppName("ReqMediaAnalysis")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    //创建sparkContext
    val sparkContext = new SparkContext(sparkConf)
    //将字段广播出去
    val appdictMap = sparkContext.textFile(ConfigHandler.appdictPath)
      //对每条数据进行切分，并过滤掉长度小于5的数据
      .map(line => line.split("\t",-1))
      .filter(_.length>=5)
      //提取appid，appname并组装成元组
      .map(arr => (arr(4),arr(1)))
      //将数据收集到Driver端，并将转换成map
      .collect().toMap
    val appdictBT = sparkContext.broadcast(appdictMap)
    //创建sqlContext实例
    val sQLContext = new SQLContext(sparkContext)
    //读取数据
    val rawDataFrame = sQLContext.read.parquet(ConfigHandler.parquetPath)
    import sQLContext.implicits._
    val result = rawDataFrame.filter("appid!=null or appid!='' or appname!=null or appname!=''")
      .map(row =>{
        val appId = row.getAs[String]("appid")
        var appName = row.getAs[String]("appname")
        if (StringUtils.isEmpty(appName)){
           appName = appdictBT.value.getOrElse(appId,appId)
        }
        (appName,RptKpiTools.offLineKpi(row))
      })
      //偏函数
      .reduceByKey{
      case (list1,list2) =>list1 zip list2 map(tp => tp._1 + tp._2)
    }.map(rs =>ReportMediaAnalysis(rs._1,rs._2(0),rs._2(1),rs._2(2),rs._2(3),rs._2(4),rs._2(5),rs._2(6),rs._2(7),rs._2(8)))
      .toDF()
    MysqlHandler.save2db(result,ConfigHandler.mediatable)

    sparkContext.stop()
  }
}
